package org.qloudgen.borg.messaging;

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.ibatis.session.SqlSession;
import org.hornetq.api.core.TransportConfiguration;  
import org.hornetq.api.core.client.*;  
import org.hornetq.core.remoting.impl.invm.InVMConnectorFactory;
import org.hornetq.jms.client.HornetQConnectionFactory;
import org.hornetq.jms.client.HornetQQueue;

import org.qloudgen.borg.CommonValues;
import org.qloudgen.borg.messaging.message.TaskMessage;
import org.qloudgen.borg.metadb.DefaultImpl;
import org.qloudgen.borg.task.TaskRules;
import org.qloudgen.borg.thread.Worker;

public class HornetqJmsClient  
{  
	static Log log = LogFactory.getLog(HornetqJmsClient.class);
	
	private String queueName ;
	private MessageProducer producer;
	private MessageConsumer consumer;
	private Session session;
	private HornetQQueue queue;
	private Connection connection;
	private SqlSession dbsess;
    
	
	public HornetqJmsClient(String queueName){
		
		this.setQueueName(queueName);
		this.setQueue(new HornetQQueue(this.getQueueName()));  
        ServerLocator serverLocator = HornetQClient.createServerLocatorWithoutHA(new TransportConfiguration(InVMConnectorFactory.class.getName()));
        HornetQConnectionFactory cf = new HornetQConnectionFactory(serverLocator);  
        
        try {
			connection = cf.createConnection();
			this.setSession(connection.createSession(false, Session.AUTO_ACKNOWLEDGE));  
			this.setConnection(connection);
			connection.start();

		} catch (JMSException e) {
			log.error(e.getMessage());
		}  
       
        
	}

	public void generateProducer(){
		
		try
	      {
	         this.setProducer(this.getSession().createProducer(this.getQueue()));   
	         
	      }catch (Exception e)
	      {
	    	  log.error(e.getMessage());
	      }
	}
	
	public void generateConsumer(){
		
		try {
			 MessageConsumer messageConsumer = this.getSession().createConsumer(queue);  
			 this.dbsess = CommonValues.factory.openSession(true);
			 
			 messageConsumer.setMessageListener(new MessageListener(){

				public void onMessage(Message message) {
					
					try {
						
						if (((ObjectMessage)message).getObject().getClass() == TaskMessage.class){
							TaskMessage task =  (TaskMessage)((ObjectMessage)message).getObject();
							
							// wait till the task can be executed
							while (!TaskRules.isExecuted(task)){
								
								try {
									Thread.sleep(CommonValues.DEFAULT_SLEEP_TIME + 1000);
								} catch (InterruptedException e) {
									log.error(e.getMessage());
								}
								
							}
							// update task status
							DefaultImpl.updateTaskStatus(task,CommonValues.TASK_STATUS_OUTQUEUE);
							
							// execute task by new thread
							Worker worker = new Worker(task);
							worker.start();
							
						}
						
					} catch (JMSException e) {
						log.error(e.getMessage());
					} 
					
				}

	         });
			 
			this.setConsumer(messageConsumer);
			
		} catch (JMSException e) {
			log.error(e.getMessage());
			this.dbsess.close();
		}   
		
	}
	
	public void sendMessage(TaskMessage message){
		
		try {
			
			if (this.getProducer() == null)
				CommonValues.throwException("no producer handler,please call generateProducer()");

	        this.producer.send(this.session.createObjectMessage(message));
	        
		} catch (Exception e) {
			log.error(e.getMessage());
		}  
	}

	public void close(){
		
		try {
			this.getSession().close();
			this.getConnection().close();
			this.dbsess.close();
			
		} catch (JMSException e) {
			log.error(e.getMessage());
		}
		
	}
	
	public void setQueueName(String queueName) {
		this.queueName = queueName;
	}

	public String getQueueName() {
		return queueName;
	}

	public void setProducer(MessageProducer producer) {
		this.producer = producer;
	}

	public MessageProducer getProducer() {
		return producer;
	}

	public void setConsumer(MessageConsumer consumer) {
		this.consumer = consumer;
	}

	public MessageConsumer getConsumer() {
		return consumer;
	}

	public void setConnection(Connection connection) {
		this.connection = connection;
	}

	public Connection getConnection() {
		return connection;
	}

	private void setSession(Session session) {
		
		this.session = session;
		
	}
	
	private Session getSession() {
		
		return this.session;
		
	}

	private void setQueue(HornetQQueue queue) {
	
		this.queue = queue;
		
	}
	
	private HornetQQueue getQueue(){
		return this.queue;
	}

}
